package com.kafka.web;

import com.alibaba.fastjson.JSON;
import com.kafka.entity.Order;
import org.apache.kafka.clients.admin.AdminClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;

/** 
 * @author: LCG
 * @date: 2022-04-15 14:37:39
 * @description:  
 */
@RestController
public class KafkaTestController {

    private static SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @Autowired
    private KafkaAdmin kafkaAdmin;

   @RequestMapping("send/{value}")
   public void testSend(@PathVariable("value") String value){

       /*ProducerRecord<String,String> producerRecord=new ProducerRecord<String,String>(
               "TEST-TOPIC",System.currentTimeMillis(),value,value);*/
       kafkaTemplate.send("TEST-TOPIC",UUID.randomUUID().toString(),value);

   }

    @RequestMapping("send2/{value}")
    public void testSend2(@PathVariable("value") String value){

       /*ProducerRecord<String,String> producerRecord=new ProducerRecord<String,String>(
               "TEST-TOPIC",System.currentTimeMillis(),value,value);*/
        String clockSequence = ""+UUID.randomUUID().toString();
        kafkaTemplate.send("MYTOPIC",clockSequence,value);

    }


    /**
     *@author: licigao
     *@date: 
     *@description: 
     *
     *@return
     */
    @RequestMapping("admin")
    public void testKafkaAdmin(){

        System.out.println(kafkaAdmin);

        AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfig());


    }


    /** 
     * @author: LCG
     * @date: 2022-06-28 12:50:23
     * @description:  测试Flink，模拟发送订单数据，统计订单每天的GMV
     */
    @RequestMapping("send3/{orderStatus}/{orderPrice}/{orderId}/{userId}")
    public void testSend3(@PathVariable("orderStatus") String orderStatus,
                          @PathVariable("orderPrice") Double orderPrice,
                          @PathVariable("orderId") String orderId,
                          @PathVariable("userId") String userId){

       /*ProducerRecord<String,String> producerRecord=new ProducerRecord<String,String>(
               "TEST-TOPIC",System.currentTimeMillis(),value,value);*/
        Order order=new Order();
        order.setOrderId(orderId);
        order.setOrderPrice(orderPrice);
        order.setOrderStatus(orderStatus);
        order.setUserId(userId);
        Date date = new Date();
        order.setDate(simpleDateFormat.format(date));
        order.setOrderDate(date.getTime());

        kafkaTemplate.send("TEST-TOPIC",UUID.randomUUID().toString(), JSON.toJSONString(order));

    }


    @RequestMapping("send4/{orderStatus}/{orderPrice}/{orderId}/{userId}/{items}")
    public void testSend4(@PathVariable("orderStatus") String orderStatus,
                          @PathVariable("orderPrice") Double orderPrice,
                          @PathVariable("orderId") String orderId,
                          @PathVariable("userId") String userId,
                          @PathVariable("items") Integer items
                          ){

       /*ProducerRecord<String,String> producerRecord=new ProducerRecord<String,String>(
               "TEST-TOPIC",System.currentTimeMillis(),value,value);*/
        for (int i=0;i<items;i++){
            Order order=new Order();
            order.setOrderId(orderId);
            order.setOrderPrice(orderPrice);
            order.setOrderStatus(orderStatus);
            order.setUserId(userId);
            Date date = new Date();
            order.setDate(simpleDateFormat.format(date));
            order.setOrderDate(date.getTime());

            kafkaTemplate.send("TEST-TOPIC",UUID.randomUUID().toString(), JSON.toJSONString(order));
        }


    }
    
}
